package com.dreamlin.app.common.exception.rxweaver.core

import io.reactivex.*
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.functions.Consumer
import io.reactivex.functions.Function
import org.reactivestreams.Publisher

class GlobalErrorTransformer<T>(
    private val upStreamSchedulerProvider: Suppiler<Scheduler>,
    private val downStreamSchedulerProvider: Suppiler<Scheduler>,
    private val globalOnNextRetryInterceptor: Function<T, Observable<T>>,
    private val globalOnErrorResume: Function<Throwable, Observable<T>>,
    private val retryConfigProvider: Function<Throwable, RetryConfig>,
    private val globalDoOnErrorConsumer: Consumer<Throwable>
) : ObservableTransformer<T, T>, FlowableTransformer<T, T>, SingleTransformer<T, T>,
    MaybeTransformer<T, T>, CompletableTransformer {

    constructor(
        globalOnNextRetryInterceptor: Function<T, Observable<T>>,
        globalOnErrorResume: Function<Throwable, Observable<T>>,
        retryConfigProvider: Function<Throwable, RetryConfig>,
        globalDoOnErrorConsumer: Consumer<Throwable>
    ) : this(
        SCHEDULER_PROVIDER_DEFAULT,
        SCHEDULER_PROVIDER_DEFAULT,
        globalOnNextRetryInterceptor,
        globalOnErrorResume,
        retryConfigProvider,
        globalDoOnErrorConsumer
    )

    override fun apply(upstream: Observable<T>): ObservableSource<T> {
        return upstream
            .observeOn(upStreamSchedulerProvider.call())
            .flatMap { t -> globalOnNextRetryInterceptor.apply(t) }
            .onErrorResumeNext(Function<Throwable, ObservableSource<out T>> { throwable ->
                globalOnErrorResume.apply(
                    throwable
                )
            })
            .retryWhen(ObservableRetryDelay(retryConfigProvider))
            .doOnError(globalDoOnErrorConsumer)
            .observeOn(downStreamSchedulerProvider.call())
    }

    override fun apply(upstream: Completable): CompletableSource {
        return upstream
            .observeOn(upStreamSchedulerProvider.call())
            .onErrorResumeNext { throwable ->
                globalOnErrorResume.apply(throwable)
                    .ignoreElements()
            }
            .retryWhen(FlowableRetryDelay(retryConfigProvider))
            .doOnError(globalDoOnErrorConsumer)
            .observeOn(downStreamSchedulerProvider.call())
    }

    override fun apply(upstream: Flowable<T>): Publisher<T> {
        return upstream
            .observeOn(upStreamSchedulerProvider.call())
            .flatMap { t ->
                globalOnNextRetryInterceptor.apply(t)
                    .toFlowable(BackpressureStrategy.BUFFER)
            }
            .onErrorResumeNext(Function<Throwable, Publisher<T>> { throwable ->
                globalOnErrorResume.apply(throwable)
                    .toFlowable(BackpressureStrategy.BUFFER)
            })
            .retryWhen(FlowableRetryDelay(retryConfigProvider))
            .doOnError(globalDoOnErrorConsumer)
            .observeOn(downStreamSchedulerProvider.call())
    }

    override fun apply(upstream: Maybe<T>): MaybeSource<T> {
        return upstream
            .observeOn(upStreamSchedulerProvider.call())
            .flatMap { t ->
                globalOnNextRetryInterceptor.apply(t)
                    .firstElement()
            }
            .onErrorResumeNext(Function<Throwable, MaybeSource<T>> { throwable ->
                globalOnErrorResume.apply(throwable)
                    .firstElement()
            })
            .retryWhen(FlowableRetryDelay(retryConfigProvider))
            .doOnError(globalDoOnErrorConsumer)
            .observeOn(downStreamSchedulerProvider.call())
    }

    override fun apply(upstream: Single<T>): SingleSource<T> {
        return upstream
            .observeOn(upStreamSchedulerProvider.call())
            .flatMap { t ->
                globalOnNextRetryInterceptor.apply(t)
                    .firstOrError()
            }
            .onErrorResumeNext { throwable ->
                globalOnErrorResume.apply(throwable)
                    .firstOrError()
            }
            .retryWhen(FlowableRetryDelay(retryConfigProvider))
            .doOnError(globalDoOnErrorConsumer)
            .observeOn(downStreamSchedulerProvider.call())
    }

    companion object {

        private val SCHEDULER_PROVIDER_DEFAULT = object : Suppiler<Scheduler> {
            override fun call(): Scheduler {
                return AndroidSchedulers.mainThread()
            }
        }
    }
}